package com.guandata.cdc.sink.impl;

import com.guandata.cdc.helper.ParamsHelper;
import com.guandata.cdc.model.sink.SinkConnectParam;
import com.guandata.cdc.model.db_type.gauss.GaussSinkConfig;
import com.guandata.cdc.model.DataTableType;
import com.guandata.cdc.sink.GuanDataSinkFactory;
import com.guandata.cdc.model.source.ConnectOptionsLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// Gauss dws cdc source https://bbs.huaweicloud.com/blogs/439795
// DWS connect https://support.huaweicloud.com/intl/en-us/tg-dws/dws_07_0179.html
@Slf4j
public class GaussSQLSinkFactory extends GuanDataSinkFactory<GaussSinkConfig> {
    public GaussSQLSinkFactory(GaussSinkConfig config, Schema schema) {
        super(config, schema);
    }

    @Override
    public String registerSinkTable(StreamTableEnvironment tableEnv, String tableName) {
        var sinkTableName = sinkTableNameStrategy.getSinkTableName();
        var sinkFlinkTableName = ParamsHelper.getInnerTableName(tableName);
        var physicalTableName = String.format("%s.%s", config.sinkSchema, sinkTableName);
        var descriptorBuilder = TableDescriptor.forConnector(config.connector).schema(this.schema);

        log.info("{} table's schema: {}", sinkFlinkTableName, this.schema);
        log.info("register physical sink table {}", physicalTableName);
        log.info("table's jdbc url: {}", config.sinkJdbcUrl);

        ConnectOptionsLoader
            .<SinkConnectParam>builder(DataTableType.GAUSS, descriptorBuilder)
            .add(SinkConnectParam.USERNAME, config.sinkUsername)
            .add(SinkConnectParam.PASSWORD, config.sinkPassword)
            .add(SinkConnectParam.TABLE_NAME, physicalTableName)
            .add(SinkConnectParam.URL, config.sinkJdbcUrl)
            .add(SinkConnectParam.AUTO_FLUSH_BATCH_SIZE, String.valueOf(50000));

        tableEnv.createTemporaryTable(sinkFlinkTableName, descriptorBuilder.build());
        return sinkFlinkTableName;
    }
}
